package org.cancer.service

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode}
import org.cancer.util.SparkUtil
import org.apache.spark.sql.functions._

//需求：4.女性患乳腺癌和宫颈癌的年龄段分析（分析高发年龄段）
class CancersFemaleService$Laurel {

  def dataAnalysis(data: DataFrame): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._ //隐式转换


    //患者编号,性别,年龄,省份,省份人口数（万人）,肿瘤类型,癌症分期,肿瘤大小,转移,治疗类型,治疗疗程/次数,生存状态,吸烟状态,饮酒习惯,基因突变,合并症
    //1,女,69,河南,9937,结直肠癌,III,6.6,是,靶向治疗+放疗+化疗,8,已故,从不,从不饮酒,APC,无

    //  数据预处理
    val filterdDF: DataFrame = data
      .filter($"性别" === "女") // 筛选女性患者
      .filter($"肿瘤类型".isin("乳腺癌", "宫颈癌")) // 筛选乳腺癌和宫颈癌
      .withColumn("年龄段",
        concat(floor($"年龄" / 10) * 10,
          lit("~"),
          floor($"年龄" / 10) * 10 + 9)
      )
/*
    // 创建组合键并统计数量
    val resultDF = filterdDF
      .withColumn("组合键", concat($"肿瘤类型", lit("_"), $"年龄段"))
      .withColumn("值", lit(1))
      .groupBy("组合键", "肿瘤类型", "年龄段")
      .agg(count("值").as("患者数量"))
      //.orderBy($"肿瘤类型", desc("患者数量"))
      .select("肿瘤类型","年龄段","患者数量")
 */
// 统计患者数量（逻辑不变）
val resultDF = filterdDF
  .groupBy("肿瘤类型", "年龄段") // 直接按肿瘤类型、年龄段分组（无需组合键）
  .agg(count("*").as("患者数量")) // 更简洁的计数方式

    // 关键：按「年龄段数值」排序（从 10~19 到 80~89）
    val sortedDF = resultDF
      .orderBy(
        // 提取年龄段的起始值（如 10~19 提取 10），转整数排序
        substring_index($"年龄段", "~", 1).cast("int")
      )

    //存入数据库
    sortedDF.write
      .format("jdbc") //指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") //链接本地 jdbc:mysql://localhost:3306/ 本地的数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") //mysql 8.0以上：com.mysql.cj.jdbc.Driver
      .option("user", "root") //用户名
      .option("password", "123456") //密码
      .option("dbtable", "female_laurel") //表名
      .mode(SaveMode.Append) //当数据不存在的时候会自动创建
      .save()

  }
}
